package io.kimmking.rpcfx.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;

import java.net.URI;
import java.util.concurrent.TimeUnit;

/**
 * 短连接客户端，进行socket通信
 */
public class ShortLinkClient {

    private ShortLinkClient() {

    }

    /**
     * TCP通信，发送并接收消息
     *
     * @param host           ip
     * @param port           端口
     * @param msg            消息
     * @param connectTimeOut 连接超时时间
     * @return 服务端返回的消息
     * @throws Exception 通信异常时，抛出此异常
     */
    public static String sendMessage(String host, int port, final String msg, int connectTimeOut, final long readTimeout) throws Exception {
        final NettyClientHandler nettyClientHandler = new NettyClientHandler(msg);
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeOut)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline p = ch.pipeline();
                            p.addLast(new HttpClientCodec());
                            p.addLast(new HttpObjectAggregator(1024 * 1024));
                            p.addLast(new HttpContentDecompressor());
                            p.addLast(new IdleStateHandler(readTimeout, 0, 0, TimeUnit.MILLISECONDS));
                            p.addLast(nettyClientHandler);
                        }
                    });

            ChannelFuture f = b.connect(host, port).sync();

            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }

        return nettyClientHandler.getResponse();
    }

    /**
     * TCP通信，发送并接收消息
     *
     * @param host    ip
     * @param port    端口
     * @param msg     消息
     * @return 服务端返回的消息
     * @throws Exception 通信异常时，抛出此异常
     */
    public static String sendMessage(String host, int port, final String msg) throws Exception {
        return sendMessage(host, port, msg, 5000, 5000);
    }

    private static class NettyClientHandler extends ChannelInboundHandlerAdapter {

        private String msg;

        private String response;

        public NettyClientHandler(String msg) {
            this.msg = msg;
        }

        public String getResponse() {
            return response;
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            URI uri = new URI("/");
            FullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_0, HttpMethod.POST,
                    uri.toASCIIString(), Unpooled.wrappedBuffer(msg.getBytes("UTF-8")));

            request.headers().add(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
            request.headers().add(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
            request.headers().add(HttpHeaderNames.CONTENT_LENGTH, request.content().readableBytes());
            ctx.writeAndFlush(request);
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof FullHttpResponse) {
                FullHttpResponse response = (FullHttpResponse) msg;
                ByteBuf buf = response.content();
                this.response = buf.toString(CharsetUtil.UTF_8);
            }
        }

        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.close();
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
            throw new Exception(cause.getMessage());
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent e = (IdleStateEvent) evt;
                if (e.state() == IdleState.READER_IDLE) {
                    ctx.close();
                }
            }
        }
    }
}
